package org.cancer.service

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.cancer.util.SparkUtil

class CancerTypeComplicationService_Faye {
  def cancerTypeComplication(data: DataFrame): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._

    val cancerComorbidityDF = data.select("肿瘤类型", "合并症")

    val cancerComorbidityRDD: RDD[(String, Array[String])] = cancerComorbidityDF.rdd.map { row =>
      val cancerType = row.getAs[String]("肿瘤类型")
      val comorbidities = row.getAs[String]("合并症").split(",")
      (cancerType, comorbidities)
    }


    val comorbidityPairs = cancerComorbidityRDD.flatMap { case (cancerType, comorbidities) =>
      comorbidities
        .filter(_ != "无")
        .map(comorbidity => (cancerType, comorbidity))
    }

    val comorbidityCounts = comorbidityPairs
      .map { case (cancerType, comorbidity) => ((cancerType, comorbidity), 1) }
      .reduceByKey(_ + _)
      .map { case ((cancerType, comorbidity), count) => (cancerType, comorbidity, count) }
      .sortBy(record => (record._1, -record._3))

    val resultDF: DataFrame = comorbidityCounts.toDF("cancer_type", "comorbidity", "count")

    resultDF.show()

    resultDF.write
      .format("jdbc") //指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") //链接本地 jdbc:mysql://localhost:3306/ 本地的数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") //mysql 8.0以上：com.mysql.cj.jdbc.Driver
      .option("user", "root") //用户名
      .option("password", "123456") //密码
      .option("dbtable", "comorbidity_counts") //表名
      .mode(SaveMode.Append) //当数据不存在的时候会自动创建
      .save()

    val queryDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://node1:3306/cancer_patients")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .option("user", "root")
      .option("password", "123456")
      .option("query", "SELECT cancer_type, comorbidity, count FROM comorbidity_counts ORDER BY cancer_type, count DESC")
      .load()

    queryDF.write
      .format("jdbc") //指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") //链接本地 jdbc:mysql://localhost:3306/ 本地的数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") //mysql 8.0以上：com.mysql.cj.jdbc.Driver
      .option("user", "root") //用户名
      .option("password", "123456") //密码
      .option("dbtable", "comorbidity_counts_faye") //表名
      .mode(SaveMode.Append) //当数据不存在的时候会自动创建
      .save()

    spark.stop()
  }
}
